package com.alibaba.nacos.naming.consistency.ephemeral.distro.v2;

import com.alibaba.nacos.common.notify.Event;
import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.common.notify.listener.MultiSubscriber;
import com.alibaba.nacos.common.utils.JacksonUtils;
import com.alibaba.nacos.consistency.DataOperation;
import com.alibaba.nacos.core.distributed.distro.DistroProtocol;
import com.alibaba.nacos.core.distributed.distro.component.DistroDataProcessor;
import com.alibaba.nacos.core.distributed.distro.component.DistroDataStorage;
import com.alibaba.nacos.core.distributed.distro.entity.DistroData;
import com.alibaba.nacos.core.distributed.distro.entity.DistroKey;
import com.alibaba.nacos.naming.core.v2.ServiceManager;
import com.alibaba.nacos.naming.core.v2.client.Client;
import com.alibaba.nacos.naming.core.v2.client.ClientSyncData;
import com.alibaba.nacos.naming.core.v2.client.manager.ClientManager;
import com.alibaba.nacos.naming.core.v2.event.client.ClientEvent;
import com.alibaba.nacos.naming.core.v2.event.client.ClientEvent.ClientChangeEvent;
import com.alibaba.nacos.naming.core.v2.event.client.ClientEvent.ClientDisconnectEvent;
import com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent;
import com.alibaba.nacos.naming.core.v2.event.metadata.MetadataEvent;
import com.alibaba.nacos.naming.core.v2.event.publisher.NamingEventPublisherFactory;
import com.alibaba.nacos.naming.core.v2.pojo.InstancePublishInfo;
import com.alibaba.nacos.naming.core.v2.pojo.Service;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;

@Slf4j
@Component
@RequiredArgsConstructor
public class DistroClientDataProcessor
        extends MultiSubscriber
        implements DistroDataStorage, DistroDataProcessor {
    public static final String TYPE = "Nacos:Naming:v2:ClientData";
    private final ClientManager clientManager;
    private final DistroProtocol distroProtocol;
    private final ServiceManager serviceManager;

    @PostConstruct
    void init() {
        NotifyCenter.registerSubscriber(this, NamingEventPublisherFactory.INSTANCE);
    }

    @Override
    public List<Class<? extends Event>> subscribeTypes() {
        return List.of(ClientChangeEvent.class, ClientDisconnectEvent.class, ClientEvent.ClientVerifyFailedEvent.class);
    }

    @Override
    public void onEvent(Event event) {
        log.info("DistroClientDataProcessor.onEvent starting event:{}", event.getClass().getCanonicalName());
        if (event instanceof ClientEvent.ClientVerifyFailedEvent) {
            throw new UnsupportedOperationException();
        }

        ClientEvent clientEvent = (ClientEvent) event;
        Client client = clientEvent.getClient();
        if (client == null || !client.isEphemeral() || !clientManager.isResponsibleClient(client)) {
            return;
        }

        DistroKey distroKey = new DistroKey(client.getClientId(), TYPE);
        distroProtocol.sync(distroKey, event instanceof ClientChangeEvent ? DataOperation.CHANGE : DataOperation.DELETE);
    }

    @Override
    public DistroData generateDistroData(DistroKey distroKey) {
        Client client = clientManager.getClient(distroKey.getResourceKey());
        if (client == null) {
            return null;
        }

        byte[] jsonBytes = JacksonUtils.toJsonBytes(client.generateSyncData());
        return new DistroData().setDistroKey(distroKey).setContent(jsonBytes);
    }

    @Override
    public boolean processData(DistroData distroData) {
        switch (distroData.getType()) {
            case ADD, CHANGE:
                ClientSyncData clientSyncData = JacksonUtils.toObj(distroData.getContent(), ClientSyncData.class);
                handlerClientSyncData(clientSyncData);
                return true;
            case DELETE:
                return true;
            default:
                return false;
        }
    }

    private void handlerClientSyncData(ClientSyncData clientSyncData) {
        clientManager.syncClientConnected(clientSyncData.getClientId(), clientSyncData.getAttributes());
        Client client = clientManager.getClient( clientSyncData.getClientId());
        upgradeClient(client, clientSyncData);
    }

    private void upgradeClient(Client client, ClientSyncData clientSyncData) {
        Set<Service> syncedService = new HashSet<>();

        List<String> namespaces = clientSyncData.getNamespaces();
        List<String> groupNames = clientSyncData.getGroupNames();
        List<String> serviceNames = clientSyncData.getServiceNames();
        List<InstancePublishInfo> instances = clientSyncData.getInstancePublishInfos();

        for (int i = 0; i < namespaces.size(); i++) {
            Service service = Service.newService(namespaces.get(i), groupNames.get(i), serviceNames.get(i));
            Service usingService = serviceManager.getOrCreate(service);
            syncedService.add(usingService);
            InstancePublishInfo instancePublishInfo = instances.get(i);
            if (!Objects.equals(instancePublishInfo, client.getInstancePublishInfo(usingService))) {
                client.addServiceInstance(usingService, instancePublishInfo);
                NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(usingService, client.getClientId()));
                NotifyCenter.publishEvent(new MetadataEvent.InstanceMetadataEvent(usingService, instancePublishInfo.getMetadataId(), false));
            }
        }

        for (Service service : client.getAllPublishedService()) {
            if (!syncedService.contains(service)) {
                client.removeServiceInstance(service);
                NotifyCenter.publishEvent(new ClientOperationEvent.ClientDeregisterServiceEvent(service, client.getClientId()));
            }
        }
        client.setRevision(clientSyncData.getAttributes().getClientAttribute("revision", 0));
    }
}
